package dev.ideas.funkyj.sequence;

import dev.ideas.funkyj.function.F;
import dev.ideas.funkyj.support.Iterators;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;

/**
 */
public class Executions {
    public static <X> F<_Sequence<X>, Iterator<X>> parallelAllExecution(final int cores) {
        return new F<_Sequence<X>, Iterator<X>>() {
            @Override public Iterator<X> apply(final _Sequence<X> xs) {
                ExecutorService executorService = Executors.newFixedThreadPool(cores);
                final ExecutorCompletionService<X> completionService = new ExecutorCompletionService<X>(executorService);

                final Iterator<?> from = xs.from.iterator();
                final List<Future<X>> futures = new LinkedList<Future<X>>();
                while (from.hasNext()) {
                    final Object next1 = from.next();

                    final Future<X> future = completionService.submit(new Callable<X>() {
                        public X call() throws Exception {
                            //noinspection unchecked
                            return (X) xs.f.apply(next1);
                        }
                    });

                    futures.add(future);
                }

                final List<X> result = new ArrayList<X>();

                for (Future<X> future : futures) {
                    try {
                        final X x = future.get();
                        if (x != null) { result.add(x); }
                    } catch (Exception e) {
                        throw new RuntimeException("", e);
                    }
                }

                return result.iterator();
            }
        };
    }

    public static <X> F<_Sequence<X>, Iterator<X>> simpleLazyExecution() {
        return new F<_Sequence<X>, Iterator<X>>() {
            @Override public Iterator<X> apply(final _Sequence<X> xs) {
                return new Iterators.FilterIterator<X>(xs.from.iterator(), xs.f);
            }
        };
    }
}
